package drds.global_transaction.jdbc;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Joiner;
import drds.common.$;
import drds.common.Author;
import drds.dbutils.native_sql.Db;
import drds.dbutils.tools.DbUtils;
import drds.global_transaction.Authors;
import drds.global_transaction.BranchTransactionInfo;
import drds.global_transaction.GlobalTransactionContext;
import drds.global_transaction.image.UndoLog;
import drds.global_transaction.image.image_praser.ImagePraser;
import drds.global_transaction.image.image_praser.InsertIImagePraser;
import drds.global_transaction.image.image_praser.SqlPraser;
import drds.global_transaction.lock.ExclusiveLock;
import drds.global_transaction.lock.Lock;
import drds.global_transaction.lock.ShareILock;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectBody;
import org.apache.commons.lang3.StringUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
@Author(name = Authors.LI_YANG)
public class GlobalTransactionPreparedStatementInvocationHandler implements InvocationHandler
{

    List<String> updateMethods = Arrays.asList("executeUpdate", "execute", "executeBatch", "executeLargeBatch", "executeLargeUpdate");
    private Object realStatementObject;
    private String sql;
    @Setter
    @Getter
    private BranchTransactionInfo branchTransactionInfo;
    private List<Object> params = new ArrayList<Object>();
    //private long timeOut = 2 * 1000;

    public GlobalTransactionPreparedStatementInvocationHandler(Object realObject, String sql)
    {
        if (GlobalTransactionContext.isInitting())
        {
            throw new RuntimeException("系统初始化中不能对外进行服务");
        }
        //
        this.realStatementObject = realObject;
        this.sql = sql;

    }


    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable
    {

        BranchTransactionInfo branchTransactionInfo = null;
        if (branchTransactionInfo == null || "close".equals(method.getName()))
        {
            return method.invoke(realStatementObject, args);
        }
        //
        if (method.getName().startsWith("set") && args != null && args.length == 2)
        {
            params.add(args[1]);
        }
        if (!branchTransactionInfo.isRunnable())
        {
            throw new SQLException(
                    "ILock operation is disabled during the inactive phase of the transaction when transaction is stoped!");
        }

        if (isUpdateMethod(method.getName()))
        {
            return invokUpdate(branchTransactionInfo, method, args);
        }

        if (isQueryMethod(method))
        {

            try
            {
                if (args == null || StringUtils.isEmpty(args[0].toString()))
                {
                    //select x
                    return method.invoke(realStatementObject, args);
                }
                sql = args[0].toString();
                if (StringUtils.deleteWhitespace(sql.toLowerCase()).endsWith("lockinsharemode"))
                {
                    return shareLockMode(method, args);
                } else
                {
                    net.sf.jsqlparser.statement.Statement statement;
                    try
                    {
                        statement = CCJSqlParserUtil.parse(sql);
                    } catch (JSQLParserException e)
                    {
                        throw new RuntimeException(e);
                    }
                    Select select = (Select) statement;
                    SelectBody selectBody = select.getSelectBody();
                    if (selectBody instanceof PlainSelect)
                    {
                        PlainSelect plainSelect = (PlainSelect) selectBody;
                        if (plainSelect.isForUpdate())
                        {
                            return lockForUpdate(method, args);
                        }
                    }

                }
            } finally
            {

            }
        }
        return method.invoke(realStatementObject, args);

    }


    private boolean isUpdateMethod(String methodName)
    {
        return updateMethods.contains(methodName);
    }

    private boolean isQueryMethod(Method method)
    {
        return "executeQuery".equals(method.getName());
    }

    private Object lockForUpdate(Method method, Object[] args)
            throws SQLException, JSQLParserException, IllegalAccessException, InvocationTargetException
    {
        UndoLog undoLog = new UndoLog();
        ImagePraser imagePraser = ImagePraser.newImagePraser(sql, undoLog);
        undoLog.setBeforeImage(imagePraser.generateBeforeImage());
        getExclusiveLockList(branchTransactionInfo, imagePraser);
        return method.invoke(realStatementObject, args);
    }

    //lock in share mode
    private Object shareLockMode(Method method, Object[] args)
            throws SQLException, JSQLParserException, IllegalAccessException, InvocationTargetException
    {
        UndoLog undoLog = new UndoLog();
        ImagePraser imagePraser = ImagePraser.newImagePraser(sql.substring(0, sql.toLowerCase().indexOf("lock")), undoLog);
        undoLog.setBeforeImage(imagePraser.generateBeforeImage());
        getShareLock(branchTransactionInfo, imagePraser);
        return method.invoke(realStatementObject, args);
    }

    private Object invokUpdate(BranchTransactionInfo branchTransactionInfo, Method method, Object[] args)
            throws SQLException, JSQLParserException, IllegalAccessException, InvocationTargetException
    {
        UndoLog undoLog = new UndoLog();
        Object object = null;
        Object primaryKeyColumnValue = null;
        if (realStatementObject instanceof PreparedStatement)
        {
            sql = sqlWithParamters(sql, params);
        } else if (realStatementObject instanceof Statement)
        {
            sql = args[0].toString();
        }
        ImagePraser imagePraser;
        List<Lock> lockedLockList;

        imagePraser = ImagePraser.newImagePraser(sql, undoLog);
        undoLog.setBeforeImage(imagePraser.generateBeforeImage());
        lockedLockList = getExclusiveLockList(branchTransactionInfo, imagePraser);
        if (SqlPraser.insert(sql))
        {
            ResultSet resultSet = null;
            if (realStatementObject instanceof PreparedStatement)
            {
                if (!branchTransactionInfo.isRunnable())
                {

                    if (lockedLockList.size() > 0)
                    {
                        releaseLockList(lockedLockList);
                    }
                    throw new SQLException(
                            "ILock operation is disabled during the inactive phase of the transaction when transaction is stoped!");
                } else
                {
                    object = method.invoke(realStatementObject, args);
                    PreparedStatement preparedStatement = (PreparedStatement) realStatementObject;
                    resultSet = preparedStatement.getGeneratedKeys();

                }

            } else if (realStatementObject instanceof Statement)
            {
                Statement statement = (Statement) realStatementObject;
                if (!branchTransactionInfo.isRunnable())
                {
                    if (lockedLockList.size() > 0)
                    {
                        releaseLockList(lockedLockList);
                    }
                    throw new SQLException(
                            "ILock operation is disabled during the inactive phase of the transaction when transaction is stoped!");
                } else
                {
                    object = statement.executeUpdate(sql, Statement.RETURN_GENERATED_KEYS);
                    resultSet = statement.getGeneratedKeys();
                }

            }
            if (resultSet != null)
            {
                while (resultSet.next())
                {
                    primaryKeyColumnValue = resultSet.getObject(1);
                }
                if (!resultSet.isClosed())
                {
                    resultSet.close();
                }
            }

            if (primaryKeyColumnValue == null)
            {
                String primaryKeyColumnName = imagePraser.getPrimaryKeyColumnName(SqlPraser.insertSentenceTableName(sql));
                List<String> insertSentenceColumnNameList = SqlPraser.insertSentenceColumnNameList(sql);
                List<String> insertSentenceColumnValueList = SqlPraser.insertSentenceColumnValueList(sql);
                if (insertSentenceColumnNameList.contains(primaryKeyColumnName))
                {
                    primaryKeyColumnValue = insertSentenceColumnValueList.get(insertSentenceColumnNameList.indexOf(primaryKeyColumnName));
                }
            }
        } else
        {
            if (!branchTransactionInfo.isRunnable())
            {
                releaseLockList(lockedLockList);
                throw new SQLException("ILock operation is disabled during the inactive phase of the transaction!");
            } else
            {
                object = method.invoke(realStatementObject, args);

            }
        }

        //插入时需要获取主键的value
        if (SqlPraser.insert(sql))
        {
            InsertIImagePraser insertImagePraser = (InsertIImagePraser) imagePraser;
            insertImagePraser.setPrimaryKeyColumnValue(primaryKeyColumnValue);
            undoLog.setAfterImage(insertImagePraser.generateAfterImage());
        } else
        {
            undoLog.setAfterImage(imagePraser.generateAfterImage());
        }
        String undoLogJson = JSON.toJSONString(undoLog);
        String logSql = "INSERT INTO global_transaction_undo_log (gmt_create,gmt_modified,gloabal_transaction_id,branch_transaction_id,rollback_info,status,server) VALUES(now(),now(),?,?,?,?,?)";
        if (object != null)
        {
            if (!branchTransactionInfo.isRunnable())
            {

                if ($.isHasData(lockedLockList))
                {
                    releaseLockList(lockedLockList);
                }
                try
                {
                    undoLog.rollback();
                } catch (SQLException e)
                {
                    log.error("UndoLog.rollback(),error", e);
                    throw e;
                }
            } else
            {
                Connection connection = null;
                PreparedStatement preparedStatement = connection.prepareStatement(logSql);
                preparedStatement.setObject(1, branchTransactionInfo.getGlobalTransactionId());
                preparedStatement.setObject(2, branchTransactionInfo.getBranchTransactionId());
                preparedStatement.setObject(3, undoLogJson);
                preparedStatement.setObject(4, 0);
                preparedStatement.setObject(5, getDatabaseUrl());
                preparedStatement.executeUpdate();

            }
        }
        //事务数据源从对应数据库获取后置对象
        return object;
    }

    private String getDatabaseUrl() throws SQLException
    {
        Connection connection = null;
        DatabaseMetaData databaseMetaData = connection.getMetaData();
        String url = databaseMetaData.getURL();
        String host = "";
        Pattern pattern = Pattern.compile("(?<=//|)((\\w)+\\.)+\\w+");
        Matcher matcher = pattern.matcher(url);
        if (matcher.find())
        {
            host = matcher.group();
        }
        DbUtils.close(connection);
        return host;
    }

    private List<Lock> getExclusiveLockList(@NonNull BranchTransactionInfo branchTransactionInfo, ImagePraser imagePraser) throws JSQLParserException, SQLException
    {


        while ((!branchTransactionInfo.isTimeOut()) && branchTransactionInfo.isRunnable())
        {
            List<Lock> lockList = this.getExclusiveLock(branchTransactionInfo, imagePraser);
            if (this.lock(lockList))
            {
                return lockList;
            }
            try
            {
                Thread.sleep(100);
            } catch (InterruptedException e)
            {
            }
        }
        throw new SQLException("Proxy.getLockTimeout");

    }

    private void getShareLock(BranchTransactionInfo branchTransactionInfo, ImagePraser imagePraser)
            throws JSQLParserException, SQLException
    {

        do
        {
            List<Lock> lockList = getShareLock(branchTransactionInfo.getGlobalTransactionId(), branchTransactionInfo.getBranchTransactionId(), imagePraser, sql);
            if (lock(lockList))
            {
                return;
            }
        } while (!branchTransactionInfo.isTimeOut());
        throw new SQLException("Proxy.getLockTimeout");
    }

    private boolean lock(List<Lock> lockList)
    {
        if (lockList.size() > 0)
        {
            for (Lock lock : lockList)
            {
                try
                {
                    if (!lock.isLock())
                    {
                        lock.lock();
                    }
                } catch (Exception e)
                {
                    log.info("getExclusiveLockList -- Data locked by other,retry");
                    return false;
                }
                lock.setLock(true);
            }
        }
        return true;
    }

    private boolean releaseLockList(List<Lock> lockList) throws SQLException
    {
        if (lockList.size() > 0)
        {
            for (Lock lock : lockList)
            {
                try
                {
                    if (lock.isLock())
                    {
                        lock.unlock();
                    }
                } catch (SQLException e)
                {
                    log.error("release error", e);
                    throw e;
                }
                lock.setLock(false);
            }
        }
        return true;
    }

    private List<Lock> getExclusiveLock(BranchTransactionInfo branchTransactionInfo, ImagePraser imagePraser)
            throws JSQLParserException, SQLException
    {

        String beforeLockSql = imagePraser.getBeforeOrAfterImageSql();
        String primaryKeyColumnName = imagePraser.getPrimaryKeyColumnName(imagePraser.getTableName());

        List<String> allLockList = new ArrayList<String>();
        ResultSet resultSet = Db.query(null, beforeLockSql);
        while (resultSet.next())
        {
            allLockList.add(resultSet.getString(primaryKeyColumnName));
        }
        List<Lock> lockList = new ArrayList<>();

        if (allLockList.size() == 0)
        {
            return lockList;
        }
        //
        String lockSql = "select primary_key_column_value from global_transaction_lock where global_transaction_id=? and branch_transaction_id =? and table_name = ? and key_value in(?)";

        String primary_key_column_values = Joiner
                .on(",")
                .skipNulls()
                .join(allLockList);
        List<Object> objectList = new ArrayList<Object>();
        objectList.add(branchTransactionInfo.getGlobalTransactionId());
        objectList.add(branchTransactionInfo.getBranchTransactionId());
        objectList.add(imagePraser.getTableName());
        objectList.add(primary_key_column_values);
        //
        Connection connection = null;
        List<String> lockedLockList = new ArrayList<String>();
        resultSet = Db.query(connection, lockSql, objectList.toArray());
        while (resultSet.next())
        {
            allLockList.add(resultSet.getString("primary_key_column_value"));
        }
        resultSet.close();
        allLockList.removeAll(lockedLockList);
        for (String primaryKeyColumnValue : allLockList)
        {

            ExclusiveLock exclusiveLock = new ExclusiveLock();
            exclusiveLock.setLock(Boolean.FALSE);
            exclusiveLock.setGlobalTransactionId(branchTransactionInfo.getGlobalTransactionId());
            exclusiveLock.setBranchTransactionId(branchTransactionInfo.getBranchTransactionId());
            exclusiveLock.setTableName(imagePraser.getTableName());
            exclusiveLock.setExclusiveLock("1");
            exclusiveLock.setShareLock(0);
            exclusiveLock.setPrimaryKeyColumnValue(primaryKeyColumnValue);
            exclusiveLock.setCreateDateTimeLong(System.currentTimeMillis());
            lockList.add(exclusiveLock);
        }

        return lockList;
    }

    private List<Lock> getShareLock(String globalTransactionId, String branchTransactionId, ImagePraser imagePraser, String sql)
            throws JSQLParserException, SQLException
    {
        String beforeLockSql = imagePraser.getBeforeOrAfterImageSql();
        String primaryKeyColumnName = imagePraser.getPrimaryKeyColumnName(imagePraser.getTableName());
        List<String> allLockList = new ArrayList<String>();
        Connection connection = null;
        ResultSet resultSet = Db.query(connection, beforeLockSql);
        while (resultSet.next())
        {
            allLockList.add(resultSet.getString(primaryKeyColumnName));
        }
        resultSet.close();
        List<Lock> lockList = new ArrayList<>();

        String lockSql =
                "select primary_key_column_value,count(*) as count from global_transaction_lock where global_transaction_id=?  and branch_transaction_id =? and table_name = ? and primary_key_column_value in(?) group by primary_key_column_value";

        String primary_key_column_values = Joiner
                .on(",")
                .skipNulls()
                .join(allLockList);
        List<Object> objectList = new ArrayList<Object>();
        objectList.add(branchTransactionInfo.getGlobalTransactionId());
        objectList.add(branchTransactionInfo.getBranchTransactionId());
        objectList.add(imagePraser.getTableName());
        objectList.add(primary_key_column_values);

        List<String> lockedLockList = new ArrayList<String>();
        resultSet = Db.query(connection, lockSql, objectList.toArray());
        while (resultSet.next())
        {
            if (resultSet.getInt("count") > 1)
            {
                lockedLockList.add(resultSet.getString("primary_key_column_value"));
            }
        }
        resultSet.close();
        allLockList.removeAll(lockedLockList);
        for (String primaryKeyColumnValue : allLockList)
        {

            ShareILock shareLock = new ShareILock();
            shareLock.setLock(Boolean.FALSE);
            shareLock.setGlobalTransactionId(globalTransactionId);
            shareLock.setBranchTransactionId(branchTransactionId);
            shareLock.setTableName(imagePraser.getTableName());
            shareLock.setExclusiveLock("1");
            shareLock.setShareLock(1);
            shareLock.setPrimaryKeyColumnValue(primaryKeyColumnValue);
            shareLock.setCreateDateTimeLong(System.currentTimeMillis());
            lockList.add(shareLock);
        }

        return lockList;
    }

    private String sqlWithParamters(String sql, List<Object> objectList) throws SQLException
    {
        if ($.isHasData(objectList))
        {
            return sql;
        }
        if (!match(sql, objectList))
        {
            log.error("SQL 语句中的占位符与参数个数不匹配,sql:" + sql);
            throw new SQLException();
        }
        int size = objectList.size();
        Object[] values = new Object[size];
        objectList.toArray(values);
        for (int i = 0; i < size; i++)
        {
            Object value = values[i];
            if (value instanceof String || value instanceof Date || value instanceof Timestamp || value instanceof Blob)
            {
                values[i] = "'" + value + "'";
            } else if (value instanceof Boolean)
            {
                values[i] = (Boolean) value ? 1 : 0;
            }
        }
        return String.format(sql.replaceAll("\\?", "%s"), values);
    }

    /**
     * ? 和参数的实际个数是否匹配
     */
    private boolean match(String sql, List<Object> objectList)
    {
        if ($.isHasData(objectList))
        {
            return true;
        }
        Matcher matcher = Pattern.compile("(\\?)").matcher(sql);
        int count = 0;
        while (matcher.find())
        {
            count++;
        }
        return count == objectList.size();
    }
}
